package com.atguigu.app.dwd.db;

import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

//交易域下单事务事实表
/*
主表是订单明细表（最细粒度）
join 订单表
left join 活动表
left join 优惠券表
lookup join 字典表(要求主流建表时设置处理时间字段)

（left join会产生左null，等右边数据来了，就撤回变成左右，所以要用upsert kafka，并且构建动态表连接kafka的时候要指定主键）
//todo 3.创建动态表连接kafka topic_db
//todo 4.过滤出订单明细表
//todo 5.过滤出订单表
//todo 6.过滤出活动表
//todo 7.过滤出优惠卷表
//todo 8.创建动态表base_dic连接mysql的base_dic表
//todo 9.五表关联
//todo 10.创建动态表dwd_order_detail连接kafka dwd_trade_order_detail主题
//todo 11.将关联查询后的结果写入dwd_order_detail（即写入kafka dwd_trade_order_detail主题）
 */
public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        //todo 2.设置检查点和状态后端
        //        // 需要从Checkpoint或者Savepoint启动程序 需启动hdfs
//        //开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        //设置超时时间为 1 分钟
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        //设置两次重启的最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        //设置任务关闭的时候保留最后一次 CK 数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //指定从 CK 自动重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1L), Time.minutes(1L)));
//        //设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC/220828");
//        //设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //todo 3.创建动态表topic_db连接kafka topic_db
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_trade_order_detail_220828"));

        //todo 4.过滤出订单明细表
        Table orderDetailTable = tableEnv.sqlQuery("select \n" +
                "data['id'] id, \n" +
                "data['order_id'] order_id, \n" +
                "data['sku_id'] sku_id, \n" +
                "data['sku_name'] sku_name, \n" +
                "data['order_price'] order_price, \n" +
                "data['sku_num'] sku_num, \n" +
                "data['create_time'] create_time, \n" +
                "data['source_type'] source_type, \n" +
                "data['source_id'] source_id, \n" +
                "data['split_total_amount'] split_total_amount, \n" +
                "data['split_activity_amount'] split_activity_amount, \n" +
                "data['split_coupon_amount'] split_coupon_amount, \n" +
                "pt \n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_detail'\n" +

                "and type='insert'");
//        orderDetailTable.execute().print();
        //注册表
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        //todo 5.过滤出订单表
        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "data['id'] id ,\n" +
                "data['consignee'] consignee ,\n" +
                "data['consignee_tel'] consignee_tel ,\n" +
                "data['total_amount'] total_amount ,\n" +
                "data['order_status'] order_status ,\n" +
                "data['user_id'] user_id ,\n" +
                "data['payment_way'] payment_way ,\n" +
                "data['delivery_address'] delivery_address ,\n" +
                "data['order_comment'] order_comment ,\n" +
                "data['out_trade_no'] out_trade_no ,\n" +
                "data['trade_body'] trade_body ,\n" +
                "data['create_time'] create_time ,\n" +
                "data['operate_time'] operate_time ,\n" +
                "data['expire_time'] expire_time ,\n" +
                "data['process_status'] process_status ,\n" +
                "data['tracking_no'] tracking_no ,\n" +
                "data['parent_order_id'] parent_order_id ,\n" +
                "data['province_id'] province_id ,\n" +
                "data['activity_reduce_amount'] activity_reduce_amount ,\n" +
                "data['coupon_reduce_amount'] coupon_reduce_amount ,\n" +
                "data['original_total_amount'] original_total_amount ,\n" +
                "data['feight_fee'] feight_fee ,\n" +
                "data['feight_fee_reduce'] feight_fee_reduce ,\n" +
                "data['refundable_time'] refundable_time \n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_info'\n" +
                "and type='insert'");

//        orderInfo.execute().print();
//
        tableEnv.createTemporaryView("order_info",orderInfo);
//
        //todo 6.过滤出订单活动明细表
        Table orderDetailActivity = tableEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['activity_id'] activity_id,\n" +
                "data['activity_rule_id'] activity_rule_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_detail_activity'\n" +
                "and type='insert'");
//        orderDetailActivity.execute().print();
        tableEnv.createTemporaryView("order_detail_activity",orderDetailActivity);
//
        //todo 7.过滤出订单优惠卷明细表
        Table orderDetailCoupon = tableEnv.sqlQuery("select\n" +
                "data['id'] id, \n" +
                "data['order_id'] order_id, \n" +
                "data['order_detail_id'] order_detail_id, \n" +
                "data['coupon_id'] coupon_id, \n" +
                "data['coupon_use_id'] coupon_use_id, \n" +
                "data['sku_id'] sku_id, \n" +
                "data['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_detail_coupon'\n" +
                "and type='insert'");

//        orderDetailCoupon.execute().print();
        tableEnv.createTemporaryView("order_detail_coupon",orderDetailCoupon);
//
//        //todo 8.创建动态表base_dic连接mysql的base_dic表
        tableEnv.executeSql(MysqlUtil.getBaseDicDDL());
//        tableEnv.sqlQuery("select * from base_dic").execute().print();
//
        //todo 9.五表关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "o.id,\n" +
                "o.order_id,\n" +
                "o.sku_id,\n" +
                "o.sku_name,\n" +
                "o.order_price,\n" +
                "o.sku_num,\n" +
                "o.create_time,\n" +
                "o.source_type,\n" +
                "o.source_id,\n" +
                "o.split_total_amount,\n" +
                "o.split_activity_amount,\n" +
                "o.split_coupon_amount,\n" +
                "i.consignee,\n" +
                "i.consignee_tel,\n" +
                "i.total_amount,\n" +
                "i.order_status,\n" +
                "i.user_id,\n" +
                "i.delivery_address,\n" +
                "i.order_comment,\n" +
                "i.out_trade_no,\n" +
                "i.trade_body,\n" +
                "i.process_status,\n" +
                "i.tracking_no,\n" +
                "i.parent_order_id,\n" +
                "i.province_id,\n" +
                "i.activity_reduce_amount,\n" +
                "i.coupon_reduce_amount,\n" +
                "i.original_total_amount,\n" +
                "i.feight_fee,\n" +
                "i.feight_fee_reduce,\n" +
                "i.refundable_time,\n" +
                "a.activity_id,\n" +
                "a.activity_rule_id,\n" +
                "c.coupon_id,\n" +
                "c.coupon_use_id,\n" +
                "d.dic_name\n" +
                "from order_detail o\n" +
                "join order_info i on o.order_id=i.id \n" +
                "left join order_detail_activity a on a.order_detail_id=o.id\n" +
                "left join order_detail_coupon c on c.order_detail_id=o.id\n" +
                "join base_dic for system_time as of o.pt as d  on i.order_status=d.dic_code");
                //lookup join

        tableEnv.createTemporaryView("result_table",resultTable);
//
        //todo 10.创建动态表dwd_order_detail连接kafka dwd_trade_order_detail主题
        tableEnv.executeSql("create table dwd_order_detail(" +
                "id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "order_price string,\n" +
                "sku_num string,\n" +
                "create_time string,\n" +
                "source_type string,\n" +
                "source_id string,\n" +
                "split_total_amount string,\n" +//支付金额
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "consignee string,\n" +
                "consignee_tel string,\n" +
                "total_amount string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "delivery_address string,\n" +
                "order_comment string,\n" +
                "out_trade_no string,\n" +
                "trade_body string,\n" +
                "process_status string,\n" +
                "tracking_no string,\n" +
                "parent_order_id string,\n" +
                "province_id string,\n" +
                "activity_reduce_amount string,\n" +
                "coupon_reduce_amount string,\n" +
                "original_total_amount string,\n" +
                "feight_fee string,\n" +
                "feight_fee_reduce string,\n" +
                "refundable_time string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "coupon_id string,\n" +
                "coupon_use_id string,\n" +
                "dic_name string," +
                "primary key (id) not enforced)"+KafkaUtil.getKafkaUpsertSinkDDL("dwd_trade_order_detail"));
                //kafka建表一定要声明主键，因为是upsertkafka
        //todo 11.将关联查询后的结果写入dwd_order_detail（即写入kafka dwd_trade_order_detail主题）
        tableEnv.executeSql("insert into dwd_order_detail select * from result_table");

    }
}
